package com.boarsoft.message.activemq;

import java.util.Map;
import java.util.Map.Entry;

import javax.annotation.PostConstruct;
import javax.jms.ConnectionFactory;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.apache.activemq.jms.pool.AmqSessionListener;
import org.apache.activemq.jms.pool.PooledSession;
import org.apache.activemq.pool.PooledConnection;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于ActiveMQ客户端完成JMS消息的外发<br>
 * 应用可以直接使用ActiveMQ的API完成外发，也可以通过本类来完成<br>
 * 将AmqClient的实例，按MessageService接口注入Messenger即可。
 * 
 * @author Mac_J
 *
 */
public class AmqClient implements ExceptionListener {
	private static final Logger log = LoggerFactory.getLogger(AmqClient.class);

	/** */
	protected String code = "default";
	/** */
	protected boolean transacted = false;
	/** */
	protected int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
	/** */
	protected PooledConnectionFactory connectionFactory;
	/** 队列列表 */
	protected Map<String, AmqDestination> destMap;

	/** */
	protected PooledConnection connection;
	/** 一个client对应一个session */
	protected PooledSession session;
	/** */
	protected AmqSessionListener sessionListener = new AmqSessionListener();

	@PostConstruct
	public void init() {
		try {
			// 初始化时，可能因为AMQ没有启动，导致失败，需要延后处理
			this.prepare();
		} catch (JMSException e) {
			log.warn("Error on init ActiveMQ client {}", code, e);
		}
		sessionListener.setClient(this);
	}

	protected void prepare() throws JMSException {
		if (session == null) {
			if (connection == null) {
				connection = (PooledConnection) connectionFactory.createConnection();
				// connection.setClientID("ebc");
				connection.setExceptionListener(this);
				connection.start();
			}
			try {
				session = (PooledSession) connection.createSession(transacted, acknowledgeMode);
				session.addSessionEventListener(sessionListener);
			} finally {
				if (session == null) {
					this.close();
				}
			}
		}
		for (Entry<String, AmqDestination> en : destMap.entrySet()) {
			en.getValue().init(session, en.getKey());
		}
	}

	public Message recevie(String from) throws Exception {
		this.prepare(); // 防止session被close
		AmqDestination d = destMap.get(from);
		AmqConsumer c = d.getConsumer();
		return c.receive();
	}

	public void put(Message msg, String target) throws Exception {
		this.prepare(); // 防止session被close
		AmqDestination d = destMap.get(target);
		AmqProducer p = d.getProducer();
		p.send(msg);
	}

	public Session getSession() throws JMSException {
		this.prepare(); // 防止session被close
		return session;
	}

	public void close() {
		if (session != null) {
			try {
				session.close();
			} catch (JMSException e) {
				log.error("Error on close jms session", e);
			} finally {
				session = null;
			}
		}
		if (connection != null) {
			try {
				connection.close();
			} catch (JMSException e) {
				log.error("Error on close jms connection", e);
			} finally {
				connection = null;
			}
		}
	}

	@Override
	public void onException(JMSException e) {
		log.error("Hit JMSException", e);
		this.close();
	}

	public void onSessionClosed(PooledSession session) {
		session = null;
	}

	public ConnectionFactory getConnectionFactory() {
		return connectionFactory;
	}

	public void setConnectionFactory(ConnectionFactory connectionFactory) {
		this.connectionFactory = (PooledConnectionFactory) connectionFactory;
	}

	public boolean isTransacted() {
		return transacted;
	}

	public void setTransacted(boolean transacted) {
		this.transacted = transacted;
	}

	public int getAcknowledgeMode() {
		return acknowledgeMode;
	}

	public void setAcknowledgeMode(int acknowledgeMode) {
		this.acknowledgeMode = acknowledgeMode;
	}

	public void setSession(Session session) {
		this.session = (PooledSession) session;
	}

	public void setDestMap(Map<String, AmqDestination> destMap) {
		this.destMap = destMap;
	}

	public String getCode() {
		return code;
	}

	public void setCode(String code) {
		this.code = code;
	}

	public AmqSessionListener getSessionListener() {
		return sessionListener;
	}

	public void setSessionListener(AmqSessionListener sessionListener) {
		this.sessionListener = sessionListener;
	}
}